package simple;

import com.google.common.base.Splitter;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author : Ashiamd email: ashiamd@foxmail.com
 * @date : 2021/1/29 10:46 PM
 * 批处理 wordcount
 */
public class WordCountBatch {
    private static final Logger logger = LoggerFactory.getLogger(WordCountBatch.class);
    public static String inputPath = "./flink-aword-count-test/src/main/resources/hello.txt";

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 从文件中读取数据
        DataSet<String> inputDataSet = env.readTextFile(inputPath);

        // 对数据集进行处理，按空格分词展开，转换成(word, 1)二元组进行统计
        // 按照第一个位置的word分组
        // 按照第二个位置上的数据求和
        DataSet<Tuple2<String, Integer>> resultSet = inputDataSet.flatMap(MyFlatMapperFunc)
                .groupBy(0)
                .sum(1);

        resultSet.print();
    }

    public static FlatMapFunction<String, Tuple2<String, Integer>> MyFlatMapperFunc = new FlatMapFunction<>() {
        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> out) throws Exception {
            // 按空格分词
            var list = Splitter.on(" ")
                    .omitEmptyStrings()
                    .trimResults()
                    .splitToList(s);

            list.forEach(word -> {
                word = word.replaceAll("\\.", "").replaceAll("\\,", "");
                out.collect(new Tuple2<>(word, 1));
            });
        }
    };
}